package com.raylu.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.raylu.bean.{DauInfo, PageLog, PageVisitInfo}
import com.raylu.util.{MyBeanUtil, MyEsUtil, MyKafkaUtil, OffsetManagerUtil, PropertiesUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import redis.clients.jedis.Jedis

import java.lang
import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util.{Date, Properties, TimeZone, UUID}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
 *
 * Description:
 *
 * Create by lucienoz on 2021/12/11.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
object DwPageVisitApp {
	def main(args : Array[String]) : Unit = {

		/*
		将页面浏览流水数据放入ES
		 */

		val sparkConf : SparkConf = new SparkConf ().setAppName ( "dw_pagevisit_app" ).setMaster ( "local[*]" )
		val ssc = new StreamingContext ( sparkConf, Seconds ( 5 ) )
		val properties : Properties = PropertiesUtil.load ( "config.properties" )
		val topic : String = properties.getProperty ( "kafka.pagelog.topic" )
		val groupId : String = "dw_pagevisit_group"


		val offset : Map[TopicPartition, Long] = OffsetManagerUtil.getOffset ( topic, groupId )

		//防止redis记录用户首次日活，而数据在进入es时发生错误，导致该用户首次日活无法被redis拦截


		var KafkaDStream : InputDStream[ConsumerRecord[String, String]] = null
		if (offset.isEmpty && offset == null) KafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( topic, ssc, groupId )
		else KafkaDStream = MyKafkaUtil.getKafkaStreamOfConsumer ( topic, ssc, groupId, offset )

		var offsetRanges : Array[OffsetRange] = null
		val KafkaDStream2 : DStream[ConsumerRecord[String, String]] = KafkaDStream.transform { RDD =>
			offsetRanges = RDD.asInstanceOf[HasOffsetRanges].offsetRanges
			RDD
		}

		KafkaDStream2.map { record =>
			val pageLog : PageLog = JSON.parseObject ( record.value (), classOf[PageLog] )
			pageLog
		}.mapPartitions { pageLogIter =>
			//关联维度数据进行维度合并
			val jedis : Jedis = RedisUtil.getJedisFromPool ()
			jedis.select ( properties.getProperty ( "redis.dimtable.db" ).toInt )

			val pageVisits : Iterator[PageVisitInfo] = for (pageLog <- pageLogIter) yield {
				var pageVisit : PageVisitInfo = PageVisitInfo ( ar = null, ba = null, ch = null, isNew = null, md = null, mid = null, os = null, uid = null, vc = null, ts = 0L, duringTime = null, lastPageId = null, pageId = null, user_gender = null, user_age = null, province_name = null, province_iso_code = null, province_3166_2 = null, province_area_code = null, dt = null, hr = null )
				println ( "pageLog=====>"+pageLog.toString )
				MyBeanUtil.copyProperties ( pageLog, pageVisit )
				println ( "pageVisit=====>"+pageVisit.toString )
				//Dim merge
				val userInfoStr : String = jedis.get ( s"DIM:user_info:${pageLog.uid}" )
				if (userInfoStr != null && !userInfoStr.isEmpty) {
					val jSONObject : JSONObject = JSON.parseObject ( userInfoStr )
					//					var user_gender : String,
					pageVisit.user_gender = jSONObject.getString ( "gender" )
					//					var user_age : String,
					val date : String = jSONObject.getString ( "birthday" )
					if (date != null) {
						val birthday : LocalDate = LocalDate.parse(date)
						val now : LocalDate = LocalDate.now ()
						val period : Period = Period.between ( birthday, now )
						pageVisit.user_age = period.getYears.toString
					}

					//					var dt : String,
					//					var hr : String
					val simpleDateFormat : SimpleDateFormat = new SimpleDateFormat ( "yyyy-MM-dd" )
					simpleDateFormat.setTimeZone ( TimeZone.getTimeZone ( "GMT+8" ) )
					val dt : String = simpleDateFormat.format ( new Date ( pageLog.ts ) )
					pageVisit.dt = dt

					val hrDateFormat : SimpleDateFormat = new SimpleDateFormat ( "HH" )
					hrDateFormat.setTimeZone ( TimeZone.getTimeZone ( "GMT+8" ) )
					val hr : String = hrDateFormat.format ( new Date ( pageLog.ts ) )
					pageVisit.hr = hr


				}

				pageVisit
			}
			jedis.close ()

			pageVisits
		}.foreachRDD { rdd =>

			rdd.foreachPartition {
				pageVisitInfoIter =>
					var dt : String = null
					val pageVisitInfoList : List[(String, PageVisitInfo)] = pageVisitInfoIter.map {
						pageVisitInfo =>
							if (dt == null) dt = pageVisitInfo.dt
							(pageVisitInfo.mid+UUID.randomUUID(), pageVisitInfo)
					}.toList
					var flag:String = null
					if(!pageVisitInfoList.isEmpty) flag = MyEsUtil.bulkSave ( s"dws_page_visit_info_${dt}", pageVisitInfoList )
					println("ES ==> "+flag)
			}
			OffsetManagerUtil.saveOffset(topic, groupId, offsetRanges)
		}


		ssc.start ()
		ssc.awaitTermination ()



	}

}
